package com.atguigu.flink.sqlfunction;


import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;


import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo4_AggFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep2.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);
       Table table = tableEnvironment.from("t1");

        MyAvg myAvg = new MyAvg();
        MyUDAF myUDAF = new MyUDAF();
        tableEnvironment.createTemporaryFunction("myUDAF",myUDAF);
        tableEnvironment.createTemporaryFunction("myAvg",myAvg);

        //TableAPI  UDAF先group by，再使用
        table
            .groupBy($("id"))
             .select(
                 $("id"),
                 Expressions.call("myUDAF",$("vc"))
                 //Expressions.call("myAvg",$("vc"))
             )
             .execute().print();
    }

    /*
        下面几个方法是每个 AggregateFunction 必须要实现的：

            createAccumulator()
            accumulate()
                public void accumulate(ACC accumulator, [user defined inputs]);
            getValue()
     */
    public static class MyUDAF extends AggregateFunction<Integer,Integer>{

        private Integer result;
        //返回最终结果
        @Override
        public Integer getValue(Integer accumulator) {
            return result;
        }

        //创建累加器
        @Override
        public Integer createAccumulator() {
            System.out.println("MyUDAF.createAccumulator");
            result = 0;
            return result;
        }

        //累加
        public void accumulate(Integer acc,Integer vc){
            result += vc;
        }

    }

    public static class MyAvg extends AggregateFunction<Double,MyAcc>{

        public void accumulate(MyAcc myAcc,Integer vc){
            myAcc.num += 1;
            myAcc.vc += vc;

        }
        //返回最终结果
        @Override
        public Double getValue(MyAcc accumulator) {
            return accumulator.vc / accumulator.num;
        }

        //创建累加器
        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0d);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer num;
        private Double vc;
    }


}
